package com.study.flink.kafka

/**
  *
  * @author stephen
  * @create 2019-05-12 13:59
  * @since 1.0.0
  */
object KafkaFlinkStreamDemo {

  def main(args: Array[String]): Unit = {

//    //添加隐式转换,否则会报错
//    import org.apache.flink.api.scala._
//
//    val kafkaProps = new Properties()
//    //kafka的一些属性
//    kafkaProps.setProperty("bootstrap.servers", "127.0.0.1:9092")
//    //所在的消费组
//    kafkaProps.setProperty("group.id", "group-test")
//    //获取当前的执行环境
//    val env = StreamExecutionEnvironment.getExecutionEnvironment
//    //kafka的consumer，test1是要消费的topic
//    val kafkaSource = new FlinkKafkaConsumer010[String](
//      "test",
//      new SimpleStringSchema,
//      kafkaProps)
//    //设置从最新的offset开始消费
//    kafkaSource.setStartFromGroupOffsets()//setStartFromLatest()
//    //自动提交offset
//    kafkaSource.setCommitOffsetsOnCheckpoints(true)
//
//    //checkpoint的时间间隔
//    env.enableCheckpointing(5000)
//    //添加consumer
//    val stream = env.addSource(kafkaSource)
//
//    stream.print()
//
//    env.execute("Kafka Flink Stream Demo")
  }
}
